package com.wsjj.yjh.watermark;

import org.apache.commons.lang3.time.DateFormatUtils;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.Partitioner;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

public class WatermarkIdlenessDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);
//        env.enableCheckpointing(2000, CheckpointingMode.EXACTLY_ONCE);

        DataStreamSource<String> source = env.socketTextStream("hadoop102", 7777);
        DataStream<String> stringDataStream = source.partitionCustom(new Partitioner<String>() {
            @Override
            public int partition(String key, int numPartitions) {
                return Integer.parseInt(key) % numPartitions;
            }
        }, new KeySelector<String, String>() {
            @Override
            public String getKey(String value) throws Exception {
                System.out.println(value);
                return value;
            }
        });
        stringDataStream.map(Integer::parseInt)
                        .assignTimestampsAndWatermarks(WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofSeconds(3)))
                                .keyBy(value -> value % 2)
                                        .window(TumblingProcessingTimeWindows.of(Time.seconds(10)))
                                                .process(new ProcessWindowFunction<Integer, String, Integer, TimeWindow>() {
                                                    @Override
                                                    public void process(Integer integer, ProcessWindowFunction<Integer, String, Integer, TimeWindow>.Context context, Iterable<Integer> elements, Collector<String> out) throws Exception {
                                                        String start = DateFormatUtils.format(context.window().getStart(), "yyyy-MM-dd HH:mm:ss.SSS");
                                                        String end = DateFormatUtils.format(context.window().getEnd(), "yyyy-MM-dd HH:mm:ss.SSS");
                                                        int count = elements.spliterator().characteristics();

                                                        out.collect("key=" + integer + "的窗口[" + start + "," + end + ")包含" + count + "条数据===>" + elements.toString());

                                                    }
                                                })
                                                        .print();

        env.execute();
    }
}
